package com.atguigu.Flink.datastream.transform;

import com.atguigu.Flink.POJO.AppEvent;
import com.atguigu.Flink.POJO.ThirdPartyEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.security.PrivateKey;
import java.util.HashMap;
import java.util.Map;

public class Flink10_BillCheck {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> ds1 = env.socketTextStream("hadoop102", 8888);
        DataStreamSource<String> ds2 = env.socketTextStream("hadoop102", 9999);
        SingleOutputStreamOperator<AppEvent> AppEve = ds1.map(new MapFunction<String, AppEvent>() {
            @Override
            public AppEvent map(String s) throws Exception {
                String[] split = s.split(",");
                return new AppEvent(split[0].trim(), split[1].trim(), Long.valueOf(split[2].trim()));
            }
        }).filter(appEvent -> appEvent.getEventType().equals("pay"));
        ds1.print();
        SingleOutputStreamOperator<ThirdPartyEvent> ThirdEve = ds2.map(
                line -> {
                    String[] split = line.split(",");
                    return new ThirdPartyEvent(split[0].trim(), split[1].trim(), split[2].trim(), Long.valueOf(split[3].trim()));
                }

        ).filter(thirdPartyEvent -> thirdPartyEvent.getEventType().equals("pay"));
        ds2.print();
        AppEve.keyBy(AppEvent::getOrderId)
                .connect(ThirdEve.keyBy(ThirdPartyEvent::getOrderId))
                .process(new KeyedCoProcessFunction<String, AppEvent, ThirdPartyEvent, String>() {
                    //使用map维护数据
                    private Map<String,AppEvent> appMap = new HashMap<>();
                    private Map<String,ThirdPartyEvent> ThirdMap = new HashMap<>();
                    //处理app数据
                    @Override
                    public void processElement1(AppEvent appEvent, KeyedCoProcessFunction<String, AppEvent, ThirdPartyEvent, String>.Context context, Collector<String> collector) throws Exception {
                        if(ThirdMap.containsKey(appEvent.getOrderId())){
                            collector.collect(appEvent.getOrderId() + "第三方先到，app后到，对账成功");
                        }else{
                            appMap.put(appEvent.getOrderId(),appEvent);
                        }

                    }

                    //处理Third数据
                    @Override
                    public void processElement2(ThirdPartyEvent thirdPartyEvent, KeyedCoProcessFunction<String, AppEvent, ThirdPartyEvent, String>.Context context, Collector<String> collector) throws Exception {
                        if(appMap.containsKey(thirdPartyEvent.getOrderId())){
                            collector.collect(thirdPartyEvent.getOrderId() + "app先到，第三方后到，对账成功");
                        }else{
                            ThirdMap.put(thirdPartyEvent.getOrderId(),thirdPartyEvent);
                        }

                    }
                }).print();


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}
